package net.j4love.nio.channels;

import org.junit.Assert;
import org.junit.Test;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.channels.CompletionHandler;
import java.nio.channels.FileLock;
import java.nio.charset.Charset;
import java.nio.file.OpenOption;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.HashSet;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * @author he peng
 * @create 2018/6/1 17:55
 * @see
 */
public class AsynchronousFileChannelTest {

    // 异步文件通道测试 , 读 、 写 、 获取锁 的操作都可以是异步的

    @Test
    public void openTest() throws Exception {

        Path path = Paths.get("D:\\Documents\\Pictures\\test\\天净沙·秋思.txt");
        Set<OpenOption> openOptions = new HashSet<>(2);
        openOptions.add(StandardOpenOption.CREATE_NEW);
        openOptions.add(StandardOpenOption.WRITE);

        ExecutorService executor = Executors.newCachedThreadPool(tf -> new Thread(tf , "AsynchronousFileChannelTest-Thread"));
        AsynchronousFileChannel fChannel = AsynchronousFileChannel.open(path , openOptions , executor);
        Assert.assertTrue(fChannel.isOpen());

        System.out.println("async file channel is open");
    }

    @Test
    public void futureWriteTest() throws Exception {
        Path path = Paths.get("D:\\Documents\\Pictures\\test\\天净沙·秋思.txt");

        Set<OpenOption> openOptions = new HashSet<>(2);
        openOptions.add(StandardOpenOption.CREATE_NEW);
        openOptions.add(StandardOpenOption.WRITE);

        ExecutorService executor = Executors.newCachedThreadPool(tf -> new Thread(tf , "AsynchronousFileChannelTest-Thread"));
        AsynchronousFileChannel fChannel = AsynchronousFileChannel.open(path , openOptions , executor);
        Assert.assertTrue(fChannel.isOpen());

        System.out.println("async file channel is open");

        String s = "天净沙·秋思\n" +
                "元代：马致远\n" +
                "枯藤老树昏鸦，\n" +
                "小桥流水人家，\n" +
                "古道西风瘦马。\n" +
                "夕阳西下，\n" +
                "断肠人在天涯。";
        byte[] bytes = s.getBytes(Charset.forName("utf-8"));
        ByteBuffer buffer = ByteBuffer.allocateDirect(bytes.length);
        buffer.put(bytes);
        buffer.flip();

        Future<Integer> future = fChannel.write(buffer, 0);
        while (! future.isDone()) {
            if (future.isCancelled()) {
                System.out.println("write operation is cancelled");
                break;
            }
            System.out.println("wait write operation done");
        }
        System.out.println("write operation is done");
        fChannel.close();
    }

    @Test
    public void completionHandlerWriteTest() throws Exception {
        Path path = Paths.get("D:\\Documents\\Pictures\\test\\天净沙·秋思.txt");
        Set<OpenOption> openOptions = new HashSet<>(2);
        openOptions.add(StandardOpenOption.CREATE_NEW);
        openOptions.add(StandardOpenOption.WRITE);

        ExecutorService executor = Executors.newCachedThreadPool(tf -> new Thread(tf , "AsynchronousFileChannelTest-Thread"));
        AsynchronousFileChannel fChannel = AsynchronousFileChannel.open(path , openOptions , executor);
//        AsynchronousFileChannel fChannel = AsynchronousFileChannel.open(path , StandardOpenOption.CREATE_NEW , StandardOpenOption.WRITE);

        Assert.assertTrue(fChannel.isOpen());

        System.out.println("async file channel is open");

        String s = "天净沙·秋思\n" +
                    "元代：马致远\n" +
                    "枯藤老树昏鸦，\n" +
                    "小桥流水人家，\n" +
                    "古道西风瘦马。\n" +
                    "夕阳西下，\n" +
                    "断肠人在天涯。";
        byte[] bytes = s.getBytes(Charset.forName("utf-8"));
        ByteBuffer buffer = ByteBuffer.allocateDirect(bytes.length);
        buffer.put(bytes);
        buffer.flip();

        fChannel.write(buffer, 0, "write", new CompletionHandler<Integer, String>() {

            @Override
            public void completed(Integer result, String attachment) {
                System.out.println("operation completed , result -> " + result + " attachment -> " + attachment);
                System.out.println("Thread -> " + Thread.currentThread().getName());
                try {
                    fChannel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

            @Override
            public void failed(Throwable exc, String attachment) {
                System.out.println("operation fails attachment -> " + attachment);
                System.err.println("operation fails error : " + exc);
            }
        });

//        Thread.sleep(10000);
        fChannel.close();
    }

    @Test
    public void futureReadTest() throws Exception {
        Path path = Paths.get("D:\\Documents\\Pictures\\test\\天净沙·秋思.txt");
        Set<OpenOption> openOptions = new HashSet<>(2);
        openOptions.add(StandardOpenOption.CREATE_NEW);
        openOptions.add(StandardOpenOption.READ);

        ExecutorService executor = Executors.newCachedThreadPool(tf -> new Thread(tf , "AsynchronousFileChannelTest-Thread"));
        AsynchronousFileChannel fChannel = AsynchronousFileChannel.open(path , openOptions , executor);

        Assert.assertTrue(fChannel.isOpen());

        System.out.println("async file channel is open");

        ByteBuffer buffer = ByteBuffer.allocate((int) fChannel.size());

        Future<Integer> future = fChannel.read(buffer, 0);

        long startTime = System.nanoTime();
        while (! future.isDone()) {
            if (future.isCancelled()) {
                System.out.println("read operation is cancelled");
                break;
            }
            System.out.println("wait read operation done");
        }

            System.out.println("read operation is done , wait time = " + (System.nanoTime() - startTime) +  " ns");
        buffer.flip();
        byte[] bytes = new byte[buffer.remaining()];
        buffer.get(bytes);
        System.out.println("read data ---------> \n" + new String(bytes , Charset.forName("utf-8")));
        fChannel.close();
    }

    @Test
    public void syncLockTest0() throws Exception {
        Path path = Paths.get("D:\\Documents\\Pictures\\test\\天净沙·秋思.txt");
        Set<OpenOption> openOptions = new HashSet<>(2);
        openOptions.add(StandardOpenOption.CREATE);
        openOptions.add(StandardOpenOption.READ);
        openOptions.add(StandardOpenOption.WRITE);

        ExecutorService executor = Executors.newCachedThreadPool(tf -> new Thread(tf , "AsynchronousFileChannelTest-Thread"));
        AsynchronousFileChannel fChannel = AsynchronousFileChannel.open(path , openOptions , executor);

        Assert.assertTrue(fChannel.isOpen());

        System.out.println("async file channel is open");

        ByteBuffer buffer = ByteBuffer.allocate((int) fChannel.size());

        // 同步的尝试获取文件锁
        FileLock lock = fChannel.tryLock();
        if (Objects.isNull(lock) || ! lock.isValid()) {
            System.out.println("file lock is invalid");
            return;
        }
        try {
            Future<Integer> future = fChannel.read(buffer, 0);

            long startTime = System.nanoTime();
            while (! future.isDone()) {
                if (future.isCancelled()) {
                    System.out.println("read operation is cancelled");
                    break;
                }
                System.out.println("wait read operation done");
            }

            System.out.println("read operation is done , wait time = " + (System.nanoTime() - startTime) +  " ns");
            buffer.flip();
            byte[] bytes = new byte[buffer.remaining()];
            buffer.get(bytes);
            System.out.println("read data ---------> \n" + new String(bytes , Charset.forName("utf-8")));
            Thread.sleep(100000);
        } finally {
            if (Objects.nonNull(lock)) {
                lock.release();
            }
            fChannel.close();
        }
    }

    @Test
    public void syncLockTest1() throws Exception {
        Path path = Paths.get("D:\\Documents\\Pictures\\test\\天净沙·秋思.txt");
        Set<OpenOption> openOptions = new HashSet<>(2);
        openOptions.add(StandardOpenOption.CREATE);
        openOptions.add(StandardOpenOption.READ);
        openOptions.add(StandardOpenOption.WRITE);

        ExecutorService executor = Executors.newCachedThreadPool(tf -> new Thread(tf , "AsynchronousFileChannelTest-Thread"));
        AsynchronousFileChannel fChannel = AsynchronousFileChannel.open(path , openOptions , executor);

        Assert.assertTrue(fChannel.isOpen());

        System.out.println("async file channel is open");

        ByteBuffer buffer = ByteBuffer.allocate((int) fChannel.size());

        // 同步的尝试获取文件锁 , 获取锁的通道必须是可读和可写的。
        FileLock lock = fChannel.tryLock();
        if (Objects.isNull(lock) || ! lock.isValid()) {
            System.out.println("file lock is invalid");
            return;
        }
        try {
            Future<Integer> future = fChannel.read(buffer, 0);

            long startTime = System.nanoTime();
            while (! future.isDone()) {
                if (future.isCancelled()) {
                    System.out.println("read operation is cancelled");
                    break;
                }
                System.out.println("wait read operation done");
            }

            System.out.println("read operation is done , wait time = " + (System.nanoTime() - startTime) +  " ns");
            buffer.flip();
            byte[] bytes = new byte[buffer.remaining()];
            buffer.get(bytes);
            System.out.println("read data ---------> \n" + new String(bytes , Charset.forName("utf-8")));
        } finally {
            if (Objects.nonNull(lock)) {
                lock.release();
            }
            fChannel.close();
        }
    }


    @Test
    public void syncLocakTest0() throws Exception {
        Path path = Paths.get("D:\\Documents\\Pictures\\test\\天净沙·秋思.txt");
        Set<OpenOption> openOptions = new HashSet<>(3);
        openOptions.add(StandardOpenOption.CREATE);
        openOptions.add(StandardOpenOption.READ);
        openOptions.add(StandardOpenOption.WRITE);

        ExecutorService executor = Executors.newCachedThreadPool(tf -> new Thread(tf , "AsynchronousFileChannelTest-Thread"));
        AsynchronousFileChannel fChannel = AsynchronousFileChannel.open(path , openOptions , executor);

        Assert.assertTrue(fChannel.isOpen());

        System.out.println("async file channel is open");

        ByteBuffer buffer = ByteBuffer.allocate((int) fChannel.size());

        // 异步的获取文件锁
        Future<FileLock> lockFuture = fChannel.lock();
        if (Objects.isNull(lockFuture) || ! lockFuture.isDone()) {
            System.out.println("file lock future is not done");
            return;
        }
        FileLock lock = lockFuture.get();
        if (Objects.isNull(lock) || ! lock.isValid()) {
            System.out.println("file lock is invalid");
            return;
        }
        try {
            Future<Integer> future = fChannel.read(buffer, 0);

            long startTime = System.nanoTime();
            while (! future.isDone()) {
                if (future.isCancelled()) {
                    System.out.println("read operation is cancelled");
                    break;
                }
                System.out.println("wait read operation done");
            }

            System.out.println("read operation is done , wait time = " + (System.nanoTime() - startTime) +  " ns");
            buffer.flip();
            byte[] bytes = new byte[buffer.remaining()];
            buffer.get(bytes);
            System.out.println("read data ---------> \n" + new String(bytes , Charset.forName("utf-8")));
            Thread.sleep(100000);
        } finally {
            if (Objects.nonNull(lock)) {
                lock.release();
            }
            fChannel.close();
        }
    }


    @Test
    public void syncLocakTest1() throws Exception {
        Path path = Paths.get("D:\\Documents\\Pictures\\test\\天净沙·秋思.txt");
        Set<OpenOption> openOptions = new HashSet<>(3);
        openOptions.add(StandardOpenOption.CREATE);
        openOptions.add(StandardOpenOption.READ);
        openOptions.add(StandardOpenOption.WRITE);

        ExecutorService executor = Executors.newCachedThreadPool(tf -> new Thread(tf , "AsynchronousFileChannelTest-Thread"));
        AsynchronousFileChannel fChannel = AsynchronousFileChannel.open(path , openOptions , executor);

        Assert.assertTrue(fChannel.isOpen());

        System.out.println("async file channel is open");

        ByteBuffer buffer = ByteBuffer.allocate((int) fChannel.size());

        // 异步的获取文件锁
        Future<FileLock> lockFuture = fChannel.lock();
        while (Objects.isNull(lockFuture) || ! lockFuture.isDone()) {
            System.out.println("file lock future is not done");
        }
        FileLock lock = lockFuture.get();
        if (Objects.isNull(lock) || ! lock.isValid()) {
            System.out.println("file lock is invalid");
            return;
        }
        try {
            Future<Integer> future = fChannel.read(buffer, 0);

            long startTime = System.nanoTime();
            while (! future.isDone()) {
                if (future.isCancelled()) {
                    System.out.println("read operation is cancelled");
                    break;
                }
                System.out.println("wait read operation done");
            }

            System.out.println("read operation is done , wait time = " + (System.nanoTime() - startTime) +  " ns");
            buffer.flip();
            byte[] bytes = new byte[buffer.remaining()];
            buffer.get(bytes);
            System.out.println("read data ---------> \n" + new String(bytes , Charset.forName("utf-8")));
        } finally {
            if (Objects.nonNull(lock)) {
                lock.release();
            }
            fChannel.close();
        }
    }
}
